package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.bean.OrderDetail;
import com.atguigu.gmall.realtime.bean.OrderInfo;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/*
读取订单和订单明细数据 形成订单宽表
 */
public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        //TODO 0.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        // TODO 1. 从kafka的dwd层接收订单和订单明细数据
        String orderInfoSourceTopic = "dwd_order_info";
        String orderDetailSourceTopic = "dwd_order_detail";
        String orderWideSinkTopic = "dwm_order_wide";
        String groupId = "order_wide_group";

        // 从kafka中读取数据
        FlinkKafkaConsumer<String> sourceOrderInfo = MyKafkaUtil.getKafkaSource(orderInfoSourceTopic,groupId);
        FlinkKafkaConsumer<String> sourceOrderDetail =MyKafkaUtil.getKafkaSource(orderDetailSourceTopic,groupId);

        DataStream<String> orderInfoJsonDS = env.addSource(sourceOrderInfo);
        DataStream<String> orderDetailJsonDS = env.addSource(sourceOrderDetail);

        //对实时数据进行结构的转换
        DataStream<OrderInfo> orderInfoDS = orderInfoJsonDS.map(
                new RichMapFunction<String, OrderInfo>() {
                    SimpleDateFormat simpleDateFormat = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
                    }

                    @Override
                    public OrderInfo map(String s) throws Exception {
                        OrderInfo orderInfo = JSON.parseObject(s,OrderInfo.class);
                        orderInfo.setCreate_ts(simpleDateFormat.parse(orderInfo.getCreate_time()).getTime());
                        return orderInfo;
                    }
                }
        );

        DataStream<OrderDetail> orderDetailDS = orderDetailJsonDS.map(
                new RichMapFunction<String, OrderDetail>() {

                    SimpleDateFormat simpleDateFormat = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
                    }

                    @Override
                    public OrderDetail map(String s) throws Exception {
                        OrderDetail orderDetail = JSON.parseObject(s,OrderDetail.class);
                        orderDetail.setCreate_ts(simpleDateFormat.parse(orderDetail.getCreate_time()).getTime());
                        return orderDetail;
                    }
                }
        );

        //TODO: 下次需要在这里看一下输出日志是不是和数据库的数据保持一致
        orderInfoDS.print("orderInfoDS:::");
        orderDetailDS.print("orderDetailDS:::");


        //TODO 2. 设定事件时间水位
        SingleOutputStreamOperator<OrderInfo> orderInfoWithEventTimeDS =
                orderInfoDS.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderInfo>forMonotonousTimestamps().withTimestampAssigner(
                                new SerializableTimestampAssigner<OrderInfo>() {
                                    @Override
                                    public long extractTimestamp(OrderInfo orderInfo, long l) {
                                        return orderInfo.getCreate_ts();
                                    }
                                }
                        )
                );

        SingleOutputStreamOperator<OrderDetail> orderDetailWithEventTimeDS =
                orderDetailDS.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderDetail>forMonotonousTimestamps().withTimestampAssigner(
                                new SerializableTimestampAssigner<OrderDetail>() {
                                    @Override
                                    public long extractTimestamp(OrderDetail orderDetail, long l) {
                                        return orderDetail.getCreate_ts();
                                    }
                                }
                        )
                );

        // TODO 3.创建合并后的宽表实体类
        // TODO 3. 设定关联的key
        KeyedStream<OrderInfo,Long> orderInfoLongKeyedStream = orderInfoWithEventTimeDS.keyBy(
                OrderInfo::getId
        );

        KeyedStream<OrderDetail,Long> orderDetailLongKeyedStream = orderDetailWithEventTimeDS.keyBy(
                OrderDetail::getOrder_id
        );

        // TODO 4.  订单和订单明细表关联 设置正负5秒，防止在业务系统中主表与从表保存的时间差
        // TODO 4.1 流和流之间的连接,输出宽表
        SingleOutputStreamOperator<OrderWide> orderWideDS =
                orderInfoLongKeyedStream.intervalJoin(orderDetailLongKeyedStream)
                .between(Time.seconds(-5),Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context context, Collector<OrderWide> collector) throws Exception {
                        collector.collect(new OrderWide(orderInfo,orderDetail));
                    }
                });

        //TODO: 查看输出信息 关联之后的信息
        orderWideDS.print("joined ::: ");

        //TODO 5. 关联用户维度，在宽表中补充宽表信息
        /*
        异步查询需要使用的流式AsyncDataStream，
        无序等待方法
        有序等待方法
         */
        SingleOutputStreamOperator<OrderWide> orderWideWithUserDStream =
                AsyncDataStream.unorderedWait(orderWideDS,
                        new DimAsyncFunction<OrderWide>("DIM_USER_INFO") {
                            //宽表和 流对象的连接
                            @Override
                            public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                                    SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
                                    String birthday = jsonObject.getString("BIRTHDAY");
                                    Date date = formatter.parse(birthday);

                                    Long curTs= System.currentTimeMillis();
                                    Long betweenMs = curTs - date.getTime();
                                    Long ageLong = betweenMs /1000L /60L /60L /24L /365L;
                                    Integer age = ageLong.intValue();
                                    orderWide.setUser_age(age);
                                    orderWide.setUser_gender(jsonObject.getString("GENDER"));
                            }

                            @Override
                            public String getKey(OrderWide orderWide) {
                                return String.valueOf(orderWide.getUser_id());
                            }
                        },60, TimeUnit.SECONDS);

        orderWideWithUserDStream.print("dim join user: ");



        //TODO 6.关联省市维度
        SingleOutputStreamOperator<OrderWide> orderWideWithProvincDStream =
                AsyncDataStream.unorderedWait(
                        orderWideWithUserDStream, new DimAsyncFunction<OrderWide>("DIM_BASE_PROVINCE") {
                            @Override
                            public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                                orderWide.setProvince_name(jsonObject.getString("NAME"));
                                orderWide.setProvince_3116_2_code(jsonObject.getString("ISO_3166_2"));
                                orderWide.setProvince_iso_code(jsonObject.getString("ISO_CODE"));
                                orderWide.setProvince_area_code(jsonObject.getString("AREA_CODE"));
                            }

                            @Override
                            public String getKey(OrderWide orderWide) {
                                return String.valueOf(orderWide.getProvince_id());
                            }
                        },60,TimeUnit.SECONDS);

        // TODO 7. 关联SKU维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSkuDStream = AsyncDataStream.unorderedWait(
                orderWideWithProvincDStream, new DimAsyncFunction<OrderWide>("DIM_SKU_INFO") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                            orderWide.setSku_name(jsonObject.getString("SKU_NAME"));
                            orderWide.setCategory3_id(jsonObject.getLong("CATEGORY3_ID"));
                            orderWide.setSpu_id(jsonObject.getLong("SPU_ID"));
                            orderWide.setTm_id(jsonObject.getLong("TM_ID"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getSku_id());
                    }
                },60,TimeUnit.SECONDS);

        // TODO 8.关联SPU商品维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSpuDStream = AsyncDataStream.unorderedWait(
                orderWideWithSkuDStream, new DimAsyncFunction<OrderWide>("DIM_SPU_INFO") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                        orderWide.setSpu_name(jsonObject.getString("SPU_NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getSpu_id());
                    }
                },60,TimeUnit.SECONDS);

        //TODO 9. 关联品类维度
        SingleOutputStreamOperator<OrderWide> orderWideWithCategory3DDStream = AsyncDataStream
                .unorderedWait(orderWideWithSpuDStream, new DimAsyncFunction<OrderWide>("DIM_BASE_CATEGORY3") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                        orderWide.setCategory3_name(jsonObject.getString("NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getCategory3_id());
                    }
                },60,TimeUnit.SECONDS);

        //TODO 10.关联品牌维度
        SingleOutputStreamOperator<OrderWide> orderWideWithTmDStream = AsyncDataStream.unorderedWait(
                orderWideWithCategory3DDStream, new DimAsyncFunction<OrderWide>("DIM_BASE_TRADEMARK") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws Exception {
                        orderWide.setTm_name(jsonObject.getString("TM_NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getTm_id());
                    }
                },60,TimeUnit.SECONDS);

        //TODO 11 将订单和订单明细表Join之后以及维度关联的宽表写到kafka的dwm层
        orderWideWithTmDStream.map(JSON::toJSONString)
                .addSink(MyKafkaUtil.getKafkaSink(orderWideSinkTopic));

        env.execute();
    }
}
